package com.wanke.subscribe;

import cn.beepower.domain.bus.MqttDataBus;
import cn.beepower.domain.bus.ThreadMode;
import cn.beepower.eig.msg.MeasureValue;
import com.wanke.scada.mqtt.MessageKit;
import com.wanke.config.MeasurementCacheConfig;
import com.wanke.constant.MqttTopicConstants;
import com.wanke.domain.po.Measurement;
import io.vertx.core.AbstractVerticle;
import io.vertx.core.Promise;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.function.BiConsumer;

/**
 * 接受来自mqtt的信息,将其发送到eventBus上去
 */
public class MqttMeasuremenSubscribetRecevice extends AbstractVerticle implements BiConsumer<String, byte[]> {

    private MqttDataBus mqttDataBus;

    private String topic;

    private String beeId;

    private Long code;

    private static final Logger logger = LoggerFactory.getLogger(MqttMeasuremenSubscribetRecevice.class);

    private MeasurementCacheConfig measurementCacheService = MeasurementCacheConfig.INSTANCE;

    public MqttMeasuremenSubscribetRecevice(MqttDataBus mqttDataBus, Long code, String beeId) {
        this.mqttDataBus = mqttDataBus;
        this.beeId = beeId;
        this.code = code;
    }

    @Override
    public void start(Promise<Void> startPromise) throws Exception {
        topic = MqttTopicConstants.MEAS_CHANGE.getTopic() + beeId;
        mqttDataBus.subscribe(topic, 1, ThreadMode.Async, beeMessage2 -> {
            this.accept(topic, beeMessage2.getM().getPayload().getValue().toByteArray());
            return null;
        });
    }

    @Override
    public void accept(String rcvTopic, byte[] bytes) {
        MeasureValue value = new MeasureValue();
        MessageKit.mergeFrom(value, bytes);
        Measurement measureValuePO = new Measurement(value);
        measureValuePO.setPointNumber(Long.valueOf(code.toString() + measureValuePO.getPointNumber().toString()));
        measurementCacheService.insert(measureValuePO);
        logger.debug("Topic:{};PoindId:{};Value:{};time;{}", rcvTopic, measureValuePO.getPointNumber(), measureValuePO.getVal(),measureValuePO.getRtime());
    }
}


